package com.lmq
import org.apache.log4j
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

import scala.collection.mutable
class computeSim() {


}
  object computeSim {
    def main(args: Array[String]): Unit = {

      val logger: Logger = log4j.Logger.getLogger(getClass)
      logger.setLevel(Level.WARN)

      val spark: SparkSession = SparkSession.builder()
        .appName(this.getClass.getName)
        .master("local[*]")
        .getOrCreate()
      val sc: SparkContext = spark.sparkContext
      import spark.implicits._
      //  def main(args: Array[String]): Unit = {
//      def RetWH1: RDD[(String, String, Double, Double)] = {


        //    val sample_knn_top50: DataFrame = spark.read.option("header","true").csv("src/main/resources/sample_knn_top50.csv")
        val sample_knn_top50: RDD[Row] = spark.read.option("header", "true").csv("src/main/resources/sample_knn_top50.csv").rdd

        //    sample_knn_top50.printSchema()
        //    sample_knn_top50.show(false)

        val sample_buy_info: RDD[Row] = spark.read.option("header", "true").csv("src/main/resources/sample_buy_info.csv").rdd
        //    sample_buy_info.printSchema()
        //    sample_buy_info.show(false)
        //1 stand for buy, 2 for click
        val sbfilter1: RDD[(String, Iterable[String])] = sample_buy_info.filter(x => x.getString(2) == "1.0")
          .map(x => (x(0).toString, x(1).toString))
          .groupByKey()
//        sbfilter1.foreach(println)

        val sbfilter2: RDD[(String, Iterable[String])] = sample_buy_info.filter(x => x.getString(2) == "2.0")
          .map(x => (x(0).toString, x(1).toString))
          .groupByKey()

        //  sample_knn_top50.join(sbinfo).show(false)


        //    println(dataset.toString())
        var a: mutable.Map[String, Iterable[String]] = mutable.Map.empty[String, Iterable[String]]
      var sb1 = sc.broadcast(a)

        sbfilter1.foreach(xyz =>{
          sb1.value += (xyz._1 -> xyz._2)
        } )

      var b: mutable.Map[String, Iterable[String]] = mutable.Map.empty[String, Iterable[String]]
      var sb2 = sc.broadcast(b)

      sbfilter2.foreach((xyz: (String, Iterable[String])) =>{
        sb2.value += (xyz._1 -> xyz._2)
      } )
        println("=============Map==================")
//      sb2.value.foreach(println)
      println("=============Mapend==================")


        val temp4 = sample_knn_top50.map((k: Row) => {
          println("hello===>",sb1.value.getOrElse(k(0).toString, Nil).toList
            .intersect(sb1.value.getOrElse(k(1).toString, Nil).toList))
          val buyOver1 = sb1.value.getOrElse(k(0).toString, Nil).toList
            .intersect(sb1.value.getOrElse(k(1).toString, Nil).toList).length * 2
          val buyDown1 = sb1.value.getOrElse(k(0).toString, Nil).toList
            .union(sb1.value.getOrElse(k(1).toString, Nil).toList).distinct.length * 3
          val buyOver2 = sb2.value.getOrElse(k(0).toString, Nil).toList
            .intersect(sb2.value.getOrElse(k(1).toString, Nil).toList).length
          val buyDown2 = sb2.value.getOrElse(k(0).toString, Nil).toList
            .union(sb2.value.getOrElse(k(1).toString, Nil).toList).distinct.length * 3


          (k(0).toString, k(1).toString, {
            if (buyDown1 == 0) 0.0 else buyOver1 / buyDown1
          }, {
            if (buyDown2 == 0) 0.0 else buyOver2 / buyDown2
          })

        })
//            temp4.filter((x: (String, String, Double, Double)) => x._4 + x._3>0.0).foreach(println)
//            temp4.foreach(println)
            temp4


      }
    }
//  }
